package com.naja.event.tracking.extract.service.impl;

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.naja.event.tracking.common.constant.RedisKeyConstant;
import com.naja.event.tracking.extract.domain.TBaseEvent;
import com.naja.event.tracking.extract.domain.TBizEvent;
import com.naja.event.tracking.extract.domain.TEventTraceAggregated;
import com.naja.event.tracking.extract.observer.Base2bizExtractObserver;
import com.naja.event.tracking.extract.repository.mybatis.TEventTraceAggregatedDao;
import com.naja.event.tracking.extract.service.TBaseEventService;
import com.naja.event.tracking.extract.service.TBizEventService;
import com.naja.event.tracking.extract.service.TEventTraceAggregatedService;
import org.redisson.api.RBucket;
import org.redisson.api.RedissonClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.ObjectFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import java.text.SimpleDateFormat;
import java.util.*;

/**
 * 埋点事件链路聚合(TEventTraceAggregated)表服务实现类
 *
 * @author EasyCode
 * @since 2023-09-18 10:02:25
 */
@Service()
public class TEventTraceAggregatedServiceImpl extends ServiceImpl<TEventTraceAggregatedDao, TEventTraceAggregated> implements TEventTraceAggregatedService {
    private final static Logger logger = LoggerFactory.getLogger(TEventTraceAggregatedServiceImpl.class);
    @Autowired
    private TBaseEventService tBaseEventService;
    @Autowired
    private TBizEventService tBizEventService;
    @Autowired
    private ObjectFactory<Map<String, Base2bizExtractObserver>> observerMapFactory;

    @Autowired
    private RedissonClient redissonClient;

    @Override
    public void incrExtractBaseEventToBiz(Date beginCreateDate, String... observers) {
        Date now = new Date();
        RBucket<Date> bucket = null;
        if (beginCreateDate == null) {
            bucket = redissonClient.getBucket(RedisKeyConstant.LATEST_TASK_EXEC_TIME_KEY);
            beginCreateDate = bucket.get();
            if(beginCreateDate==null){
                beginCreateDate = now;
            }
        }
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        String formatBeginDate = simpleDateFormat.format(beginCreateDate);
        logger.debug("beginCreateDate => {}", formatBeginDate);
        int offset = 0;
        int limit = 1000;
        List<TEventTraceAggregated> tEventTraceAggregateds = null;
        Map<String, Base2bizExtractObserver> observerMap = observerMapFactory.getObject();
        if (observers != null && observers.length > 0) {
            Map<String, Base2bizExtractObserver> tmp = observerMap;
            observerMap = new HashMap<>();
            for (String observer : observers) {
                Base2bizExtractObserver base2bizExtractObserver = tmp.get(observer);
                if (base2bizExtractObserver != null) {
                    observerMap.put(observer, base2bizExtractObserver);
                }
            }
        }
        if (observerMap.isEmpty()) {
            logger.warn("选择的观察者为空,方法返回");
            return;
        }
        do {
            tEventTraceAggregateds = this.baseMapper.pageNewsByCreateTime(formatBeginDate, offset, limit);
            offset += limit;
            List<TBizEvent> allBizEvent = new ArrayList<>();
            for (TEventTraceAggregated aggr : tEventTraceAggregateds) {
                logger.info("处理单条TEventTraceAggregated=>{}", JSONObject.toJSONString(aggr));
                try {
                    this.handleSimpleTEventTraceAggregated(aggr, observerMap, allBizEvent);
                } catch (Exception e) {
                    logger.error("处理单条TEventTraceAggregated[{}]异常捕获", JSONObject.toJSONString(aggr), e);
                }finally {
                    for (Base2bizExtractObserver value : observerMap.values()) {
                        value.init();
                    }
                }
            }
            if (!allBizEvent.isEmpty()) {
                this.tBizEventService.saveBatch(allBizEvent);
            }
        } while (tEventTraceAggregateds.size() == limit);
        if (bucket != null) {
            bucket.set(now);
        }
    }

    private void handleSimpleTEventTraceAggregated(TEventTraceAggregated aggr, Map<String, Base2bizExtractObserver> observerMap, List<TBizEvent> allBizEvent) {
        if (!StringUtils.isEmpty(aggr.getTraceId()) && aggr.getActionBeginTime() != null && aggr.getActionEndTime() != null) {
            List<TBaseEvent> simpleTraceBaseEvent = this.tBaseEventService.listByTraceIdAndActionTimeRange(aggr.getTraceId(), aggr.getActionBeginTime(), aggr.getActionEndTime());
            for (TBaseEvent event : simpleTraceBaseEvent) {
                observerMap.forEach((key, value) -> {
                    try {
                        value.update(aggr, event);
                    } catch (Exception e) {
                        logger.error("观察者[{}]处理单条TBaseEvent[{}]异常捕获", JSONObject.toJSONString(value), JSONObject.toJSONString(event), e);
                    }
                });
            }
            observerMap.forEach((key, value) -> {
                try {
                    List<TBizEvent> result = value.getResult();
                    if (!result.isEmpty()) {
                        allBizEvent.addAll(result);
                    }
                } catch (Exception e) {
                    logger.error("循环获取单条TEventTraceAggregated[{}]观察者结果集异常捕获", JSONObject.toJSONString(aggr), e);
                } finally {
                    value.init();
                }
            });
        }
    }
}

